package amqp

import (
	"fmt"
	"giot/pkg/logger"
	amqp "github.com/rabbitmq/amqp091-go"
)

// mqUrl amqp://user:password@host:port/vhost
// amqp://是固定参数，这个信息是固定不变的。后面两个是用户名密码ip地址端口号Virtual Host
// 如果vhost是“/”就输入/%2F，/%2F代表斜杠
//var mqUrl = "amqp://guest:guest@127.0.0.1:5672/"

// RabbitMQ rabbitMQ结构体
type RabbitMQ struct {
	conn    *amqp.Connection
	channel *amqp.Channel
	//队列名称
	queueName string
	//接收队列
	acceptName string
	//交换机名称
	exchange string
	//bind key 名称
	key string
	//连接信息
	url string

	callback func(msg string)
}

func (r *RabbitMQ) SetMqUrl(url string) {
	r.url = url
}

func (r *RabbitMQ) SetCallback(f func(s string)) {
	r.callback = f
}

func (r *RabbitMQ) SetAcceptQueue(accept string) {
	r.acceptName = accept
}

// newRabbitMQ 创建结构体实例
func (r *RabbitMQ) newRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
	return &RabbitMQ{queueName: queueName, exchange: exchange, key: key, url: r.url}
}

// Destroy 断开channel 和 connection
func (r *RabbitMQ) Destroy() {
	if r.channel != nil {
		_ = r.channel.Close()
	}
	if r.conn != nil {
		_ = r.conn.Close()
	}
}

// 错误处理函数
func (r *RabbitMQ) failOnErr(err error, message string) {
	if err != nil {
		logger.Errorf("%s:%s", message, err)
		panic(fmt.Sprintf("%s:%s", message, err))
	}
}

// NewRabbitMQWork 创建work模式下RabbitMQ实例
func (r *RabbitMQ) NewRabbitMQWork(queueName string) *RabbitMQ {
	//创建RabbitMQ实例
	rabbitmq := r.newRabbitMQ(queueName, "", "")
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.url)
	rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
	//获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

// PublishWork work模式队列生产
func (r *RabbitMQ) PublishWork(message string) error {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	_, err := r.channel.QueueDeclare(
		// queue:队列名称
		r.queueName,
		//durable:是否持久化，当mq重启之后，还在
		true,
		//exclusive:是否独占即只能有一个消费者监听这个队列
		false,
		//autoDelete:是否自动删除。当没有Consumer时，自动删除掉
		false,
		//noWait:是否阻塞处理。true:不阻塞，false:阻塞
		false,
		//arguments:其他属性
		nil,
	)
	if err != nil {
		return err
	}
	//调用channel 发送消息到队列中
	err = r.channel.Publish(
		r.exchange,
		r.queueName,
		//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false,
		//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})

	return err

}

// ConsumeWork work 模式下消费者
func (r *RabbitMQ) ConsumeWork() error {
	//1.申请队列，如果队列不存在会自动创建，存在则跳过创建
	q, err := r.channel.QueueDeclare(
		//队列名称
		r.acceptName,
		//是否持久化
		true,
		//是否自动删除
		false,
		//是否具有排他性
		false,
		//是否阻塞处理
		false,
		//额外的属性
		nil,
	)
	if err != nil {
		return err
	}
	//接收消息
	msgs, err := r.channel.Consume(
		q.Name, // queue
		//用来区分多个消费者
		"", // consumer
		//是否自动应答
		true, // auto-ack
		//是否独有
		false, // exclusive
		//设置为true，表示 不能将同一个Conenction中生产者发送的消息传递给这个Connection中 的消费者
		false, // no-local
		//列是否阻塞
		false, // no-wait
		nil,   // args
	)
	if err != nil {
		return err
	}

	//启用协程处理消息
	go func() {
		for d := range msgs {
			//消息逻辑处理，可以自行设计逻辑
			if r.callback != nil {
				go r.callback(string(d.Body))
			}
			logger.Debugf("Received a message: %s", d.Body)
		}
	}()
	return nil
}

// NewRabbitMQRouting 路由模式 创建RabbitMQ实例
func (r *RabbitMQ) NewRabbitMQRouting(exchangeName string, routingKey string) *RabbitMQ {
	//创建RabbitMQ实例
	rabbitmq := r.newRabbitMQ("", exchangeName, routingKey)
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.url)
	rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
	//获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

// PublishRouting 路由模式发送消息
func (r *RabbitMQ) PublishRouting(message string) error {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		//交换机名称
		r.exchange,
		//交换机类型
		"direct",
		//是否持久化
		true,
		//是否自动删除
		false,
		//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		//是否阻塞处理
		false,
		//其他属性
		nil,
	)
	if err != nil {
		return err
	}
	//r.failOnErr(err, "Failed to declare an exchange")

	//2.发送消息
	//发送消息
	//参数：
	//1.交换机
	//2.队列，
	//3.是否强制性路由到队列，如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还
	//给发送者
	//4.是否立即发送
	//5.其他属性
	//6.发送消息的内容
	//调用channel 发送消息到队列中
	err = r.channel.Publish(
		r.exchange,
		r.key,
		//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false,
		//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	return err
}

// ConsumeRouting routing 模式下消费者
func (r *RabbitMQ) ConsumeRouting() error {
	//1.试探性创建交换机
	err := r.channel.ExchangeDeclare(
		// 交换机名称
		r.exchange,
		//交换机类型
		"direct",
		//是否持久化
		true,
		// 是否自动删除
		false,
		//YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		//
		false,
		nil,
	)
	if err != nil {
		return err
	}
	//r.failOnErr(err, "Failed to declare an exchange")

	//2.试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		//随机生产队列名称
		"",
		//是否持久化
		false,
		//是否自动删除
		false,
		//是否具有排他性
		true,
		//是否阻塞处理
		false,
		//其他参数
		nil,
	)
	if err != nil {
		return err
	}
	//r.failOnErr(err, "Failed to declare a queue")

	//绑定队列到 exchange 中
	err = r.channel.QueueBind(
		q.Name,
		// routing模式下要指定key
		r.key,
		r.exchange,
		false,
		nil)

	//消费消息
	messages, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)

	//启用协程处理消息
	go func() {
		for d := range messages {
			if r.callback != nil {
				go r.callback(string(d.Body))
			}
			logger.Debugf("Received a message: %s", d.Body)
		}
	}()
	return nil
}

// NewRabbitMQPub 创建Pub模式下RabbitMQ实例
func (r *RabbitMQ) NewRabbitMQPub(exchangeName string) *RabbitMQ {
	//创建RabbitMQ实例
	rabbitmq := r.newRabbitMQ("", exchangeName, "")
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.url)
	rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
	//获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

// PublishPub Pub模式队列生产
func (r *RabbitMQ) PublishPub(message string) error {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		//交换机名称
		r.exchange,
		//交换机类型
		"fanout",
		//是否持久化
		true,
		//是否自动删除
		false,
		//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		//是否阻塞处理
		false,
		//其他属性
		nil,
	)
	if err != nil {
		return err
	}
	//r.failOnErr(err, "Failed to declare an exchange")

	//2.发送消息
	//发送消息
	//参数：
	//1.交换机
	//2.队列，
	//3.是否强制性路由到队列，如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还
	//给发送者
	//4.是否立即发送
	//5.其他属性
	//6.发送消息的内容
	//调用channel 发送消息到队列中
	err = r.channel.Publish(
		r.exchange,
		"",
		//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false,
		//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	return err
}

// ConsumePub pub 模式下消费者
func (r *RabbitMQ) ConsumePub() error {
	//1.试探性创建交换机
	err := r.channel.ExchangeDeclare(
		// 交换机名称
		r.exchange,
		//交换机类型
		"fanout",
		//是否持久化
		true,
		// 是否自动删除
		false,
		//YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		//
		false,
		nil,
	)
	if err != nil {
		return err
	}
	//r.failOnErr(err, "Failed to declare an exchange")

	//2.试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		//随机生产队列名称
		"",
		//是否持久化
		false,
		//是否自动删除
		false,
		//是否具有排他性
		true,
		//是否阻塞处理
		false,
		//其他参数
		nil,
	)
	if err != nil {
		return err
	}

	//绑定队列到 exchange 中
	err = r.channel.QueueBind(
		q.Name,
		//在pub/sub模式下，这里的key要为空
		"",
		r.exchange,
		false,
		nil)
	if err != nil {
		return err
	}
	//消费消息
	messages, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return err
	}

	//启用协程处理消息
	go func() {
		for d := range messages {
			if r.callback != nil {
				go r.callback(string(d.Body))
			}
			logger.Debugf("Received a message: %s", d.Body)
		}
	}()
	return nil
}

// NewRabbitMQTopic 话题模式 创建RabbitMQ实例
func (r *RabbitMQ) NewRabbitMQTopic(exchangeName string, routingKey string) *RabbitMQ {
	//创建RabbitMQ实例
	rabbitmq := r.newRabbitMQ("", exchangeName, routingKey)
	var err error
	//获取connection
	rabbitmq.conn, err = amqp.Dial(rabbitmq.url)
	rabbitmq.failOnErr(err, "failed to connect rabbitmq!")
	//获取channel
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.failOnErr(err, "failed to open a channel")
	return rabbitmq
}

// PublishTopic 话题模式发送消息
func (r *RabbitMQ) PublishTopic(message string) error {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		//交换机名称
		r.exchange,
		//交换机类型
		"topic",
		//是否持久化
		true,
		//是否自动删除
		false,
		//true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		//是否阻塞处理
		false,
		//其他属性
		nil,
	)
	if err != nil {
		return err
	}
	//r.failOnErr(err, "Failed to declare an exchange")

	//2.发送消息
	//发送消息
	//参数：
	//1.交换机
	//2.队列，
	//3.是否强制性路由到队列，如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还
	//给发送者
	//4.是否立即发送
	//5.其他属性
	//6.发送消息的内容
	//调用channel 发送消息到队列中
	err = r.channel.Publish(
		r.exchange,
		r.key,
		//如果为true，根据自身exchange类型和routekey规则无法找到符合条件的队列会把消息返还给发送者
		false,
		//如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
	return err
}

// ConsumeTopic topic 模式下消费者
func (r *RabbitMQ) ConsumeTopic() error {
	//1.试探性创建交换机
	err := r.channel.ExchangeDeclare(
		// 交换机名称
		r.exchange,
		//交换机类型
		"topic",
		//是否持久化
		true,
		// 是否自动删除
		false,
		//YES表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		//
		false,
		nil,
	)
	if err != nil {
		return err
	}

	//2.试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		//随机生产队列名称
		"",
		//是否持久化
		false,
		//是否自动删除
		false,
		//是否具有排他性
		true,
		//是否阻塞处理
		false,
		//其他参数
		nil,
	)
	if err != nil {
		return err
	}

	//绑定队列到 exchange 中
	err = r.channel.QueueBind(
		q.Name,
		// topic模式下要指定key
		r.key,
		r.exchange,
		false,
		nil)
	if err != nil {
		return err
	}
	//消费消息
	messages, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	if err != nil {
		return err
	}

	//启用协程处理消息
	go func() {
		for d := range messages {
			if r.callback != nil {
				go r.callback(string(d.Body))
			}
			logger.Debugf("Received a message: %s", d.Body)
		}
	}()
	return nil
}
